package com.bw.fun;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.Test.DwsTradeOrgOrderDayBean;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Collections;

public class AsyncIOFuncykwide1 extends RichAsyncFunction<DwsTradeOrgOrderDayBean, DwsTradeOrgOrderDayBean> {
    private Connection conn1;
    PreparedStatement ps1 = null;
    PreparedStatement ps2 = null;


    // 从Hbase
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn1 = DriverManager.getConnection("jdbc:phoenix:hadoop102:2181");
    }

    @Override
    public void close() throws Exception {
        conn1.close();
        ps1.close();
        ps2.close();
    }

    @Override
    public void asyncInvoke(DwsTradeOrgOrderDayBean orderInfoWide, ResultFuture<DwsTradeOrgOrderDayBean> resultFuture) throws Exception {

        // 根据发件人区县ID找到机构的ID和名字
        String sql1 = "select line from DIM_BASE_ORGAN_YK9 where region_id = ?";
        ps1 = conn1.prepareStatement(sql1);
        try {
            ps1.setLong(1, Long.parseLong(orderInfoWide.getSenderDistrictId()));
            ResultSet rs1 = ps1.executeQuery();
            if (rs1.next()) {
                String line1 = rs1.getString("line");
                if (line1 != null) {
                    // 转成JSON
                    JSONObject ob1 = JSON.parseObject(line1);

                    String org_id = ob1.getString("id");
                    String org_name = ob1.getString("org_name");
                    orderInfoWide.setOrgId(org_id);
                    orderInfoWide.setOrgName(org_name);
                }
            }


            // 在管理城市的名字
            String sql2 = "select line from dim_base_region_info_yk9 where id= ?";
            ps2 = conn1.prepareStatement(sql2);
            ps2.setObject(1, orderInfoWide.getCityId());
            ResultSet rs2 = ps2.executeQuery();
            if (rs2.next()) {
                String line2 = rs2.getString("line");
                if (line2 != null) {
                    // 转成JSON
                    JSONObject ob2 = JSON.parseObject(line2);
                    String name = ob2.getString("name");
                    orderInfoWide.setCityName(name);
                }
            }
            resultFuture.complete(Collections.singleton(orderInfoWide));

        } catch (Exception e){

        }
    }
}
